package demo.connectsrc;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import demo.base.connect.OrderAddress;
import demo.base.connect.OrderDetail;

import java.util.Properties;

import scala.Tuple2;

public class FlinkConnectSourceDemo {
    public static final String KFKTPC = "test";
    public static final String KFKTPC1 = "test01";
    public static final String KFKADD = "10.10.15.243:9092,10.10.15.224:9092,10.10.15.222:9092";

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        try {
            env.setStateBackend(
                    new RocksDBStateBackend("hdfs://wetech-hdfs:8020/flink/RocksDBBackend"));

            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10000);
            ckConf.setCheckpointTimeout(60000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            Properties prop = new Properties();
            prop.setProperty("bootstrap.servers", KFKADD);
            prop.setProperty("group.id", "flkcncsrcdemo");
            prop.setProperty("auto.offset.reset", "latest");
            prop.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            prop.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            FlinkKafkaConsumer<String> kfkSource1 =
                    new FlinkKafkaConsumer<String>(KFKTPC, new SimpleStringSchema(), prop);
            KeyedStream<OrderAddress, String> orderAddressStringKeyedStream =
                    env.addSource(kfkSource1)
                            .map(new OrderAddressMapFunction())
                            .keyBy(
                                    new KeySelector<OrderAddress, String>() {
                                        @Override
                                        public String getKey(OrderAddress value) throws Exception {
                                            return value.getOrderId();
                                        }
                                    });

            FlinkKafkaConsumer<String> kfkSource2 =
                    new FlinkKafkaConsumer<String>(KFKTPC1, new SimpleStringSchema(), prop);
            KeyedStream<OrderDetail, String> orderDetailStringKeyedStream =
                    env.addSource(kfkSource2)
                            .map(new OrderDetailMapFunction())
                            .keyBy(
                                    new KeySelector<OrderDetail, String>() {
                                        @Override
                                        public String getKey(OrderDetail value) throws Exception {
                                            return value.getOrderId();
                                        }
                                    });

            //            KeyedStream<OrderDetail, String> orderDetailStringKeyedStream =
            // env.socketTextStream("localhost", 9999)
            //                    .map(new OrderDetailMapFunction())
            //                    .keyBy(new KeySelector<OrderDetail, String>() {
            //                        public String getKey(OrderDetail value) throws Exception {
            //                            return value.getOrderId();
            //                        }
            //                    });
            //            final KeyedStream<OrderAddress, String> orderAddressStringKeyedStream =
            // env.socketTextStream("localhost", 9000)
            //                    .map(new OrderAddressMapFunction())
            //                    .keyBy(new KeySelector<OrderAddress, String>() {
            //                        public String getKey(OrderAddress value) throws Exception {
            //                            return value.getOrderId();
            //                        }
            //                    });

            SingleOutputStreamOperator<Tuple2<OrderDetail, OrderAddress>> output =
                    orderDetailStringKeyedStream
                            .connect(orderAddressStringKeyedStream)
                            .flatMap(
                                    new RichCoFlatMapFunction<
                                            OrderDetail,
                                            OrderAddress,
                                            Tuple2<OrderDetail, OrderAddress>>() {
                                        private ValueState<OrderDetail> valueStateOdetail;
                                        private ValueState<OrderAddress> valueStateOdAdd;

                                        @Override
                                        public void open(Configuration parameters)
                                                throws Exception {
                                            super.open(parameters);
                                            valueStateOdetail =
                                                    getRuntimeContext()
                                                            .getState(
                                                                    new ValueStateDescriptor<
                                                                            OrderDetail>(
                                                                            "OrderDetail",
                                                                            OrderDetail.class));
                                            valueStateOdAdd =
                                                    getRuntimeContext()
                                                            .getState(
                                                                    new ValueStateDescriptor<
                                                                            OrderAddress>(
                                                                            "OrderAddress",
                                                                            OrderAddress.class));
                                        }

                                        @Override
                                        public void flatMap1(
                                                OrderDetail value,
                                                Collector<Tuple2<OrderDetail, OrderAddress>> out)
                                                throws Exception {
                                            OrderAddress odAdd = valueStateOdAdd.value();
                                            OrderDetail odetail = valueStateOdetail.value();

                                            out.collect(
                                                    new Tuple2<OrderDetail, OrderAddress>(
                                                            value,
                                                            (odAdd == null
                                                                    ? OrderAddress.empty()
                                                                    : odAdd)));
                                            if (odetail != value) {
                                                valueStateOdetail.update(value);
                                            }
                                        }

                                        @Override
                                        public void flatMap2(
                                                OrderAddress value,
                                                Collector<Tuple2<OrderDetail, OrderAddress>> out)
                                                throws Exception {
                                            OrderDetail odetail = valueStateOdetail.value();
                                            OrderAddress odAdd = valueStateOdAdd.value();

                                            out.collect(
                                                    new Tuple2<OrderDetail, OrderAddress>(
                                                            (odetail == null
                                                                    ? OrderDetail.empty()
                                                                    : odetail),
                                                            value));
                                            if (odAdd != value) {
                                                valueStateOdAdd.update(value);
                                            }
                                        }
                                    });
            output.print();

            env.execute("Flink-Connect-Source-Test");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
